/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.nio.ByteBuffer;
import java.util.logging.Logger;

import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.compression.VariableLengthUnsignedLongSequenceCodec;
import edu.ucsd.db.bassarisk.data.values.LongType;
import edu.ucsd.db.bassarisk.data.values.LongType.LongTools;
import edu.ucsd.db.bassarisk.data.values.LongType.LongValueCodec;
import edu.ucsd.db.bassarisk.data.values.LongType.ObjectLongReference;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.common.IColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.loading.ColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.reading.IColumnIndexReader.DrawerInfo;

public class DeltaEncodedAscendingUnsignedLongDrawerIndexReader implements
    IDrawerIndexReader<LongType> {
  
  private static final Logger LOG = 
      Logger.getLogger(DeltaEncodedAscendingUnsignedLongDrawerIndexReader.class.getName());
  
  protected static class Cursor {
    protected final ObjectLongReference cardEntryDelta = new ObjectLongReference();
    protected int cardIndex;
    protected long cardStartEntry;
    protected long cardStartValue;
    protected final ObjectLongReference cardValueDelta = new ObjectLongReference();
    protected int entryByteOffset;
    protected int valueByteOffset;
  }

  private static final short drawerIndexVersion = 2;
  private static final int HEADER_BYTES = 15;
  private final Cursor current;
  private int dataCardCount;
  private ByteBuffer drawerBuffer;
  private IColumnSegmentInfo<LongType> drawerInfo;
  private final VariableLengthUnsignedLongSequenceCodec longCodec;
  private final LongValueCodec longManip;
  private long segmentMaxValue;

  public DeltaEncodedAscendingUnsignedLongDrawerIndexReader() {
    this.longCodec = VariableLengthUnsignedLongSequenceCodec.INSTANCE;
    this.longManip = LongTools.INSTANCE.getGenericCodec();

    this.current = new Cursor();
  }

  // For use in test
  public DeltaEncodedAscendingUnsignedLongDrawerIndexReader(final ByteBuffer injectedBuffer,
                                                            final VariableLengthUnsignedLongSequenceCodec injectedCodec,
                                                            final Cursor injectedCursor,
                                                            final DrawerInfo<LongType> injectedDrawerInfo,
                                                            final int injectedDataCardCount,
                                                            final long injectedSegmentMaxValue) {
    this.longCodec = injectedCodec;
    this.longManip = LongTools.INSTANCE.getGenericCodec();
    this.current = injectedCursor;
    this.drawerInfo = injectedDrawerInfo;
    this.drawerBuffer = injectedBuffer;

    this.dataCardCount = injectedDataCardCount;
    this.segmentMaxValue = injectedSegmentMaxValue;
  }

  @Override
  public void getCardData(final int cardIndex, final CardInfo<LongType> info) {
    assert cardIndex >= this.current.cardIndex;
    assert cardIndex < this.dataCardCount;

    while (cardIndex > this.current.cardIndex) {
      this.stepCursor();
    }
    info.reset(this.current.cardIndex, this.current.cardEntryDelta.getValue(),
               this.current.cardStartEntry);
  }

  @Override
  public boolean getCardDataForEntryIndex(final int targetEntryIndex,
                                          final CardInfo<LongType> info) {
    assert targetEntryIndex >= this.current.cardStartEntry; // forward only

    if (!this.drawerInfo.hasEntry(targetEntryIndex)) { return false; }

    while (targetEntryIndex >= (this.current.cardStartEntry + this.current.cardEntryDelta
        .getValue())) {
      // FIXME this loop could go right past the end.
      this.stepCursor();
    }
    info.reset(this.current.cardIndex, this.current.cardEntryDelta.getValue(),
               this.current.cardStartEntry);
    return true;
  }

  @Override
  public boolean getCardDataForValue(final IValueReference<LongType> value,
                                     final CardInfo<LongType> info) {
    long longTargetValue = this.longManip.getLongValue(value);
    return this.getCardDataForValue(longTargetValue, info);
  }
  
  private long currentCardMax() {
    return this.current.cardStartValue + this.current.cardValueDelta.getValue();
  }

  public boolean getCardDataForValue(final long targetValue,
                                     final CardInfo<LongType> info) {
    assert this.current.cardIndex == -1 || 
           targetValue >= this.current.cardStartValue; // forward only iterator

    if (targetValue > this.segmentMaxValue) { return false; }
    
    if (this.current.cardIndex == -1) { this.stepCursor(); }
    while (targetValue >= this.currentCardMax()) {
      // FIXME loop goes beyond number of cards in drawer.
      this.stepCursor();
    }
    info.reset(this.current.cardIndex, this.current.cardEntryDelta.getValue(),
               this.current.cardStartEntry);
    longManip.setLongValue(this.current.cardStartValue, info.getFirstValue());
    longManip.setLongValue(this.currentCardMax(), info.getLastValue());
    
    assert targetValue >= this.current.cardStartValue;
    assert targetValue <= this.currentCardMax();
    
//    String format = "Setting CardInfo for value=%s, card_index=%s, first_entry=%s, entry_count=%s, min_value=%s, max_value=%s.";
//    LOG.info(String.format(format, 
//                           targetValue,
//                           this.current.cardIndex, 
//                           this.current.cardStartEntry, 
//                           this.current.cardEntryDelta.getValue(), 
//                           this.current.cardStartValue, 
//                           this.currentCardMax()));
    
    return true;
  }

  @Override
  public int getDataCardCount() {
    return this.dataCardCount;
  }

  @Override
  public int getDrawerIndexVersion() {
    return drawerIndexVersion;
  }

  @Override
  public boolean getNextCardData(final CardInfo<LongType> info) {
    if ((this.current.cardIndex + 1) >= this.dataCardCount) { return false; }

    this.stepCursor();
    info.reset(this.current.cardIndex, this.current.cardEntryDelta.getValue(),
               this.current.cardStartEntry);
    return true;
  }

  @Override
  public CardInfo<LongType> makeCardInfoInstance() {
    return new CardInfo<LongType>(LongTools.INSTANCE);
  }

  @Override
  public void resetDrawer(final ByteBuffer drawerBuffer,
                          final ColumnSegmentInfo<LongType> drawerInfo) {
    DrawerInfo<LongType> di = (DrawerInfo<LongType>) drawerInfo;
    int drawerHeaderAddress = di.getDrawerHeaderPageIndex();
    this.drawerInfo = drawerInfo;
    this.drawerBuffer = drawerBuffer;

    // Extract header information from the tail of the drawerBuffer
    int cardAddress = ((drawerHeaderAddress + 1) * ColumnStore.BYTES_IN_CARD);
    int headerPosition = cardAddress - HEADER_BYTES;
    this.drawerBuffer.position(headerPosition);

    int indexCardCount = this.drawerBuffer.get();
    this.dataCardCount = this.drawerBuffer.getShort();
    this.segmentMaxValue = this.drawerBuffer.getLong();
    int bytesInValueIndex = this.drawerBuffer.getShort();
    short versionRead = this.drawerBuffer.getShort();
    assert drawerIndexVersion == versionRead;
    this.drawerBuffer.clear();
    // Leave the drawerBuffer in a cleared state
    
//    String format = "Setting drawer first_entry=%s, entry_count=%s, min_value=%s, max_value=%s.";
//    LOG.info(String.format(format,
//                           this.drawerInfo.getFirstEntryIndex(), 
//                           this.drawerInfo.getEntryCount(), 
//                           this.longManip.getLongValue(this.drawerInfo.getFirstValue()),
//                           this.longManip.getLongValue(this.drawerInfo.getLastValue())));

    int firstIndexCard = drawerHeaderAddress - indexCardCount + 1;
    this.current.cardIndex = -1;
    this.current.cardStartEntry = 0;
    this.current.cardStartValue = this.longManip.getLongValue(drawerInfo.getFirstValue());
    this.longManip.setLongValue(0L, this.current.cardValueDelta);
    this.current.valueByteOffset = firstIndexCard * ColumnStore.BYTES_IN_CARD;
    this.current.entryByteOffset = this.current.valueByteOffset + bytesInValueIndex;

    this.current.entryByteOffset +=
        this.longCodec.decode(this.drawerBuffer, this.current.cardEntryDelta,
                              this.current.entryByteOffset,
                              ColumnStore.BYTES_IN_DRAWER);
  }

  private void stepCursor() {
    this.current.cardStartValue += this.current.cardValueDelta.getValue();
    this.current.cardStartEntry += this.current.cardEntryDelta.getValue();
    this.current.valueByteOffset +=
        this.longCodec.decode(this.drawerBuffer, this.current.cardValueDelta,
                              this.current.valueByteOffset,
                              ColumnStore.BYTES_IN_DRAWER);
    this.current.entryByteOffset +=
        this.longCodec.decode(this.drawerBuffer, this.current.cardEntryDelta,
                              this.current.entryByteOffset,
                              ColumnStore.BYTES_IN_DRAWER);
//    String format = "Stepping cardinfo min_value=%s, max_value=%s, value_delta=%s.";
//    LOG.info(String.format(format, 
//                           this.current.cardStartValue,
//                           this.currentCardMax(),
//                           this.current.cardValueDelta.getValue()));
    this.current.cardIndex++;
  }

}
